package com.niodata.dp.api.common;

import com.niodata.dp.collection.WrappedStringKeyMap;
import com.niodata.dp.log.LogUtil;
import com.niodata.dp.util.JsonUtil;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import okhttp3.ConnectionPool;
import okhttp3.HttpUrl;
import okhttp3.Interceptor;
import okhttp3.MediaType;
import okhttp3.OkHttpClient;
import okhttp3.Request;
import okhttp3.RequestBody;
import okhttp3.Response;
import org.apache.log4j.Logger;

/**
 * jobName Dp task plugin API.
 */
public class DpTaskApi {

  public static final MediaType JSON
        = MediaType.parse("application/json; charset=utf-8");
  private static final Logger logger = Logger.getLogger(DpTaskApi.class.getName());
  private String baseUrl = "";
  private final String server;
  private final int port;
  private final OkHttpClient client;


  /**
   * dp api.
   *
   * @param server server host
   * @param port port
   */
  public DpTaskApi(String server, int port) {
    this.server = server;
    this.port = port;
    this.baseUrl = "http://" + server + ":" + port;
    ConnectionPool pool = new ConnectionPool(2, 30, TimeUnit.SECONDS);
    OkHttpClient.Builder httpClientBuilder = new OkHttpClient.Builder()
          .connectTimeout(30, TimeUnit.SECONDS)
          .readTimeout(120L, TimeUnit.SECONDS)
          .writeTimeout(120L, TimeUnit.SECONDS)
          .retryOnConnectionFailure(true)
          .connectionPool(pool);
    Interceptor interceptor = new Interceptor() {
      @Override
      public Response intercept(Chain chain) throws IOException {
        Request request = chain.request();
        Response response = null;
        boolean success = false;
        try {
          response = chain.proceed(request);
          success = response.isSuccessful();
        } catch (IOException ioe) {
          logger.warn("request failed ");
        }
        int tryCount = 0;
        int maxLimit = 5;
        while (!success && tryCount < maxLimit) {
          logger.info("intercept:" + "Request failed - " + tryCount);
          tryCount++;
          try {
            if (response != null) {
              response.close();
            }
            Thread.sleep(2000);
          } catch (InterruptedException e) {
            LogUtil.logStackTrace(logger, e);
          }
          response = chain.proceed(request);
          if (response.code() == 404) {
            break;
          }
        }
        return response;
      }
    };
    client = httpClientBuilder.addInterceptor(interceptor).build();
  }

  public Map<String, Object> getJobConfig(String taskName, String jobName, String execId)
        throws IOException {
    Request request = new Request.Builder().url(new HttpUrl.Builder()
          .scheme("http")
          .host(server)
          .port(port)
          .addPathSegment(ApiConstants.URL_GET_JOB_CONF)
          .addQueryParameter("execId", execId)
          .addQueryParameter("taskName", taskName)
          .addQueryParameter("jobName", jobName).build()).build();
    Response response = client.newCall(request).execute();
    String body = response.body().string();
    response.close();
    return JsonUtil.fromJson(Map.class, body);
  }

  public Map<String, Object> getJobConfig(String configId) throws IOException {
    //TODO
    Request request = new Request.Builder().url(new HttpUrl.Builder()
          .scheme("http")
          .host(server)
          .port(port)
          .addPathSegment(ApiConstants.URL_GET_DYNAMIC_CONF)
          .addQueryParameter("configId", configId).build()).build();
    Response response = client.newCall(request).execute();
    String body = response.body().string();
    response.close();
    return JsonUtil.fromJson(Map.class, body);
  }

  /**
   * save job state.
   *
   * @param taskName task name
   * @param scheduleId sed id
   * @param jobName jobname
   * @param state state
   * @throws IOException e
   */
  public void saveJobState(String taskName, String scheduleId, String jobName,
        Map<String, String> state)
        throws IOException {
    Request request = new Request.Builder().url(new HttpUrl.Builder()
                .scheme("http")
                .host(server)
                .port(port)
                .addPathSegment(ApiConstants.URL_SAVE_JOB_STATE)
                .addQueryParameter("taskName", taskName)
                .addQueryParameter("jobName", jobName)
                .addQueryParameter("scheduleId", scheduleId).build())
          .post(RequestBody.create(JSON, JsonUtil.toJson(state))).build();
    Response response = client.newCall(request).execute();
    if (!response.isSuccessful()) {
      response.close();
      throw new IOException("save job state failed");
    }
    response.close();
  }

  /**
   * get job state.
   *
   * @param taskName task name
   * @param scheduleId schedule id
   * @param jobName job name
   * @return map
   * @throws IOException ex
   */
  public Map<String, Object> getJobState(String taskName, String scheduleId, String jobName)
        throws IOException {

    Request request = new Request.Builder().url(new HttpUrl.Builder()
                .scheme("http")
                .host(server)
                .port(port)
                .addPathSegment(ApiConstants.URL_GET_JOB_STATE)
                .addQueryParameter("taskName", taskName)
                .addQueryParameter("jobName", jobName)
                .addQueryParameter("scheduleId", scheduleId).build())
          .get().build();
    Response response = client.newCall(request).execute();
    if (!response.isSuccessful()) {
      response.close();
      throw new IOException("get job state failed:"
            + response.body() != null ? response.body().string() : "");
    }
    String json = response.body().string();
    response.close();
    return JsonUtil.fromJson(Map.class, json);
  }

  /**
   * save runtime config.
   *
   * @param taskName task name
   * @param jobName jobname
   * @param execId execid
   * @param configMap conf map
   * @throws IOException ioe
   */
  public void saveJobRuntimeConfig(String taskName, String jobName, String execId,
        Map<String, Object> configMap) throws IOException {

    Request request = new Request.Builder().url(new HttpUrl.Builder()
                .scheme("http")
                .host(server)
                .port(port)
                .addPathSegment(ApiConstants.URL_SAVE_JOB_RUNTIME_CONF)
                .addQueryParameter("taskName", taskName)
                .addQueryParameter("jobName", jobName)
                .addQueryParameter("execId", execId).build())
          .post(RequestBody.create(JSON, JsonUtil.toJson(configMap))).build();
    Response response = client.newCall(request).execute();
    if (!response.isSuccessful()) {
      response.close();
      throw new IOException("save job runtime config failed");
    }
    response.close();
  }

  /**
   * save output params.
   *
   * @param taskName taskName
   * @param jobName jobName
   * @param execId execid
   * @param configMap configMap
   * @throws IOException ioe
   */
  public void saveOutputParams(String taskName, String jobName, String execId,
        Map<String, Object> configMap) throws IOException {
    HttpUrl url = new HttpUrl.Builder()
          .scheme("http")
          .host(server)
          .port(port)
          .addPathSegment(ApiConstants.URL_SAVE_JOB_OUTPUT_PARAM)
          .addQueryParameter("taskName", taskName)
          .addQueryParameter("jobName", jobName)
          .addQueryParameter("execId", execId)
          .build();
    Request request = new Request.Builder().url(url).post(
          RequestBody.create(JSON, JsonUtil.toJson(configMap))
    ).build();
    Response response = client.newCall(request).execute();
    if (!response.isSuccessful()) {
      response.close();
      throw new IOException("save output params failed");
    }
    response.close();
  }


  /**
   * save output params.
   *
   * @param taskName taskName
   * @param jobName jobName
   * @param execId execid
   * @param result result
   * @throws IOException ioe
   */
  public void saveOutputParams(String taskName, String jobName, String execId,
                               Object result) throws IOException {
    HttpUrl url = new HttpUrl.Builder()
        .scheme("http")
        .host(server)
        .port(port)
        .addPathSegment(ApiConstants.URL_SAVE_JOB_OUTPUT_PARAM)
        .addQueryParameter("taskName", taskName)
        .addQueryParameter("jobName", jobName)
        .addQueryParameter("execId", execId)
        .build();
    Request request = new Request.Builder().url(url).post(
        RequestBody.create(JSON, JsonUtil.toJson(result))
    ).build();
    Response response = client.newCall(request).execute();
    if (!response.isSuccessful()) {
      response.close();
      throw new IOException("save output params failed");
    }
    response.close();
  }


  public void sendMessage(String taskName, String jobName, String execId,
        Map<String, Object> message)
        throws IOException {
    Map<String, String> params = new HashMap<>();
    params.put("taskName", taskName);
    params.put("jobName", jobName);
    params.put("execId", execId);
    Map<String, String> header = new HashMap<>();
    header.put("content-type", "application/json");
    String body = JsonUtil.toJson(message);
    String url = baseUrl + ApiConstants.URL_SEND_MESSAGE;

    //HttpCaller.post(String.class, url, params, body, header);
  }

  /**
   * get task output params.
   *
   * @param taskName task name
   * @param execId exec id
   * @return map
   * @throws IOException ex
   */
  public Map<String, Map<String, Object>> getTaskOutputParams(String taskName, String execId)
        throws IOException {

    HttpUrl url = new HttpUrl.Builder()
          .scheme("http")
          .host(server)
          .port(port)
          .addPathSegment(ApiConstants.URL_GET_TASK_OUTPUT_PARAM)
          .addQueryParameter("taskName", taskName)
          .addQueryParameter("execId", execId)
          .build();
    Request request = new Request.Builder().url(url).get().build();
    Response response = client.newCall(request).execute();
    if (!response.isSuccessful()) {
      response.close();
      throw new IOException("get output params failed");
    }
    String json = response.body().string();
    response.close();
    return JsonUtil.fromJson(Map.class, json);
  }

  /**
   * get job output params.
   *
   * @param taskName task name
   * @param jobName job name
   * @param execId exec id
   * @return map
   * @throws IOException ex
   */
  public Map<String, Object> getJobOutputParams(String taskName, String jobName, String execId)
        throws IOException {

    HttpUrl url = new HttpUrl.Builder()
          .scheme("http")
          .host(server)
          .port(port)
          .addPathSegment(ApiConstants.URL_GET_JOB_OUTPUT_PARAM)
          .addQueryParameter("taskName", taskName)
          .addQueryParameter("execId", execId)
          .addQueryParameter("jobName", jobName)
          .build();
    Request request = new Request.Builder().url(url).get().build();
    Response response = client.newCall(request).execute();
    if (!response.isSuccessful()) {
      response.close();
      throw new IOException("save output params failed");
    }
    String json = response.body().string();
    response.close();
    return JsonUtil.fromJson(Map.class, json);
  }

  /**
   * save table info.
   *
   * @param taskName taskname
   * @param dsName dsname
   * @param dbName dbname
   * @param tableName tablename
   * @param tableDesc tabledesc
   * @return success
   * @throws IOException e
   */
  public boolean saveTableInfo(String taskName, String dsName, String dbName, String tableName,
        String tableDesc)
        throws IOException {
    HttpUrl url = new HttpUrl.Builder()
          .scheme("http")
          .host(server)
          .port(port)
          .addPathSegment(ApiConstants.URL_SAVE_TABLE_INFO)
          .addQueryParameter("taskName", taskName)
          .addQueryParameter("dsName", dsName)
          .addQueryParameter("dbName", dbName)
          .addQueryParameter("tableName", tableName)
          .addQueryParameter("tableDesc", tableDesc)
          .build();
    Request request = new Request.Builder().url(url).get().build();
    Response response = client.newCall(request).execute();
    boolean succ = response.isSuccessful();
    response.close();
    return succ;
  }

  /**
   * get filesystem config by name and user.
   * @param name filesystem datasource name
   * @param user user
   * @return config
   * @throws IOException ex
   */
  public WrappedStringKeyMap getFsConfig(String name, String user) throws IOException {
    HttpUrl url = new HttpUrl.Builder()
        .scheme("http")
        .host(server)
        .port(port)
        .addPathSegment(ApiConstants.URL_GET_FS_CONFIG)
        .addQueryParameter("name", name)
        .addQueryParameter("user", user)
        .build();
    Request request = new Request.Builder().url(url).get().build();
    Response response = client.newCall(request).execute();
    if (!response.isSuccessful()) {
      response.close();
      throw new IOException("get fs config failed");
    }
    String json = response.body().string();
    response.close();
    Map map = JsonUtil.fromJson(Map.class, json);
    return new WrappedStringKeyMap(map);
  }

  /**
   * get database config by dsName and dbName.
   * @param dsName datasource name
   * @param dbName database name
   * @param user current user
   * @return config
   * @throws IOException ex
   */
  public WrappedStringKeyMap getDbConfig(String dsName, String dbName, String user) throws IOException {
    HttpUrl url = new HttpUrl.Builder()
        .scheme("http")
        .host(server)
        .port(port)
        .addPathSegment(ApiConstants.URL_GET_DB_CONFIG)
        .addQueryParameter("dsName", dsName)
        .addQueryParameter("dbName", dbName)
        .addQueryParameter("user", user)
        .build();
    Request request = new Request.Builder().url(url).get().build();
    Response response = client.newCall(request).execute();
    if (!response.isSuccessful()) {
      response.close();
      throw new IOException("get database config failed");
    }
    String json = response.body().string();
    response.close();
    Map map = JsonUtil.fromJson(Map.class, json);
    return new WrappedStringKeyMap(map);
  }
}
